package com.redis.study.stream.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;


@Slf4j
@RestController
public class MessageController {

    @Resource
    private StringRedisTemplate stringRedisTemplate;


    private StreamOperations<String, String, String> stream = null;

    @PostConstruct
    public void init(){
        LettuceConnectionFactory connectionFactory = (LettuceConnectionFactory) stringRedisTemplate.getConnectionFactory();
        // 不启用共享连接
        connectionFactory.setShareNativeConnection(false);
        stream = stringRedisTemplate.opsForStream();
    }

    /**
     *  向队列中中添加一条消息
     * @param info 消息
     * @return 消息id
     */
    @RequestMapping("/message/{info}")
    public String sendMessage(@PathVariable String info){
        Map<String,String> map = new LinkedHashMap<>();
        map.put("key",info);
        RecordId recordId = stream.add("abc", map);
        /*Map<String,String> map1 = new LinkedHashMap<>();
        map1.put("key1",info);
        RecordId recordId1 = stream.add("abc", map1);*/
        return recordId.toString();
    }
    /**
     *  获取该队列信息
     * @param key 队列key
     * @return 消息id
     */
    @RequestMapping("/group/{key}")
    public String getMessageGroup(@PathVariable String key){
        StreamInfo.XInfoGroups groups = stream.groups(key);
        StreamInfo.XInfoStream info = stream.info(key);
        log.info("groups:{}",groups.toString());
        log.info("info:{}",info.toString());
        return "ok";
    }

    /**
     *  在对应的队列中创建组
     * @param key 队列key
     * @param group 组名
     * @return 返回ok
     */
    @RequestMapping("/group/{key}/{group}")
    public String createGroup(@PathVariable String key,@PathVariable String group){
        log.info("在{}下创建了组{}",key,group);
        return stream.createGroup(key, group);
    }

    /**
     * 通key和消费组获取所有未被ack的信息  {注意：消费者不能为数字、如果是数字这块会出现转异常}
     * @param key  队列key
     * @param group 组名
     */
    @RequestMapping("/pending/{key}/{group}")
    public String getPending(@PathVariable String key,@PathVariable String group){
        PendingMessagesSummary pending = stream.pending(key, group);
        // 所有pending消息的数量
        Long totalPendingMessages = pending.getTotalPendingMessages();
        // 消费组名称
        String groupName= pending.getGroupName();
        // pending队列中的最小ID
        String minMessageId = pending.minMessageId();

        // pending队列中的最大ID
        String maxMessageId = pending.maxMessageId();

        log.info("消费组：{}，一共有{}条pending消息，最大ID={}，最小ID={}", groupName, totalPendingMessages, minMessageId, maxMessageId);

        // 查询每个消费者pending数量
        Map<String, Long> pendingMessagesPerConsumer = pending.getPendingMessagesPerConsumer();
        // 消费者名称及待消费的数据条数
        log.info(String.valueOf(pendingMessagesPerConsumer));

        return "ok";
    }

    /**
     * 获取未被ack的信息
     * @param key 队列key
     * @param group 组名
     * @param groupName 消费者
     */
    @RequestMapping("/read/{key}/{group}/{groupName}")
    public String getRead(@PathVariable String key,@PathVariable String group,@PathVariable String groupName){
        // 获取key所有的信息{堵塞状态}
//        List<MapRecord<String, String, String>> retVal1 = stream.read(StreamReadOptions.empty().block(Duration.ofMillis(5000L)), StreamOffset.fromStart(key));
        // 获取key对应组对应消费者待ack的信息
        List<MapRecord<String, String, String>> retVal = stream.read(Consumer.from(group, groupName), StreamOffset.create(key,ReadOffset.lastConsumed()));
        // 遍历消息
        for (MapRecord<String, String, String> record : retVal ) {
            // 消费消息
            log.info("消息id={}, 消息value={}", record.getId(), record.getValue());
            // 手动ack消息
//            streamOperations.acknowledge("my_group", record);
        }
        return "ok";
    }

    /**
     *  手动标记队列为ack消费状态从pending移除
     * @param key  队列key
     * @param group 组名
     * @param recordId 消息id
     * @return 消费条数
     */
    @RequestMapping("/ack/{key}/{group}/{recordId}")
    public Long ack(@PathVariable String key,@PathVariable String group,@PathVariable String recordId){
        Long acknowledge = stream.acknowledge(key, group, recordId);
        return acknowledge;
    }



}
